package com.xinqing.test.flink.processfunction;


import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.walkthrough.common.entity.Alert;
import org.apache.flink.walkthrough.common.entity.Transaction;

import java.beans.Transient;


/**
 * @Author:CHQ
 * @Date:2021/4/15 16:08
 * @Description
 */
public class FraudDetector extends KeyedProcessFunction<Long, Transaction, Alert> {
    //受多线程干扰，此处为拿到对应key的状态后端，特此设立键控状态ValueState
    private transient ValueState<Boolean> isSmallAmountflag;
    public transient ValueState<Long> timerState;
    public static final Double small_amount = 1.00;
    public static final Double large_amount = 500.00;

    //处理分区化的key
    @Override
    public void processElement(Transaction transaction, Context context, Collector<Alert> collector) throws Exception {
        //1min之内如果当前用户话费大于500元，则直接输出告警
        if (isSmallAmountflag.value() != null) {
            //消费金额大于500,立马告警
            if (transaction.getAmount() > large_amount) {
                Alert alert = new Alert();
                alert.setId(transaction.getAccountId());
                collector.collect(alert);
            }

            //清空状态后端并删除触发器
            isSmallAmountflag.clear();
            context.timerService().deleteProcessingTimeTimer(timerState.value());
            timerState.clear();
        }

        //记录当前用户小于1元的消费记录到状态后端，并启动一个1min的定时器
        if (transaction.getAmount() < small_amount) {
            isSmallAmountflag.update(true);
            timerState.update(context.timerService().currentProcessingTime() + 60 * 1000);
            context.timerService().registerProcessingTimeTimer(timerState.value());
        }
    }

    //触发器
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Alert> out) throws Exception {
        //清空状态后端
        isSmallAmountflag.clear();
        timerState.clear();
    }

    //注册状态后端
    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<Boolean> isSmallAccountflagDescriptor = new ValueStateDescriptor<>("isSmallAmountflag",
                Types.BOOLEAN);
        isSmallAmountflag = getRuntimeContext().getState(isSmallAccountflagDescriptor);

        ValueStateDescriptor<Long> timerStateDescriptor = new ValueStateDescriptor<>("timeStateDescriptor",
                Types.LONG);
        timerState = getRuntimeContext().getState(timerStateDescriptor);
    }
}

